package com.webflux.sample.webfluxmain;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import reactor.core.publisher.Flux;

public class _5_1operator_buffer {

  public static void main(String[] args) {
    // 操作符buffer和bufferTimeout 这两个操作符的作用是把当前流中的元素收集到集合中，并把集合对象作为流中的新元素。
    //在进行收集时可以指定不同的条件：所包含的元素的最大数量或收集的时间间隔。方法buffer()仅使用一个条件，而bufferTimeout()可以同时指定两个条件。
    //指定时间间隔时可以使用Duration对象或毫秒数，即使用bufferMillis()或bufferTimeoutMillis()两个方法。
    //除了元素数量和时间间隔外，还可以通过bufferUntil和bufferWhile操作符来进行收集。这两个操作符的参数时表示每个集合中的元素索要满足的条件的Predicate对象。
    //bufferUntil会一直收集直到Predicate返回true。
    //使得Predicate返回true的那个元素可以选择添加到当前集合或下一个集合中；bufferWhile则只有当Predicate返回true时才会收集。一旦为false，会立即开始下一次收集。

    System.out.println("step 1");
    Flux.range(1, 100).buffer(20).subscribe(System.out::println);

    System.out.println("step 2");
    Flux.interval(Duration.of(0, ChronoUnit.MILLIS), Duration.of(20, ChronoUnit.MILLIS)).buffer(20).take(3).toStream().forEach(System.out::println);

    System.out.println("step 3");
    Flux.range(1, 10).bufferUntil(i -> i%2 == 0).subscribe(System.out::println);

    System.out.println("step 4");
    Flux.range(1, 10).bufferWhile(i -> i%2 == 0).subscribe(System.out::println);

    System.out.println("step 5");
    Flux.interval(Duration.of(0, ChronoUnit.MILLIS), Duration.of(20, ChronoUnit.MILLIS))
        .bufferWhile(i -> i%2 == 0).buffer(20).take(3).toStream().forEach(System.out::println);

  }

}
